package cn.gwm.flink.streaming.dwd.check;

import cn.gwm.flink.streaming.beans.dwd.DwdStandardField;
import cn.gwm.flink.streaming.config.DataPrecisionConfigEnum;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.dwd.standardization.StandardFieldTrans;
import cn.gwm.flink.streaming.dwd.standardization.StandardPoolField;
import cn.gwm.flink.streaming.dwd.streamsource.CommonStreamSource;
import cn.gwm.flink.streaming.dwd.streamsource.StreamSource;
import cn.gwm.flink.streaming.sink.kafka.KafkaConsumerEnum;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.Serializable;
import java.util.List;
import java.util.Map;

/**
 * @ClassName VehicleConsumer
 * @Description
 * @Author LiangGuang
 * @Date 2023/05/18 16:05
 */
@Data
@Accessors(chain = true)
@NoArgsConstructor
public abstract class VehicleConsumer extends DwdHandleTemplate implements Serializable {

    /**
     * 需要消费的 ods层车型 主题
     */
    public String topic;
    /**
     * 消费者组
     */
    public String groupId;
    /**
     * 车型
     */
    public String vehicleType;

    /**
     * 动力类型
     */
    public String customType;
    /**
     * 里程字段，用于过滤不达标的里程车辆
     */
    public String mileage;


    public StreamSource source = new CommonStreamSource(KafkaConsumerEnum.GROUPOFFSETS, uidSuffix);

    public VehicleConsumer(String uidSuffix) {
        super(uidSuffix);
    }

    public void errorStreamHandle(DataStream<JSONObject> stream) {
        // 无任何处理
    }

    /**
     * 接收数据并进行字段处置
     *
     * @return
     */
    public DataStream<JSONObject> receiveAndHandle(StreamExecutionEnvironment env) {
        SingleOutputStreamOperator<JSONObject> mapStream = source.source(env, groupId, topic)
                .filter(item -> mileageFilter(item))
                .map(item -> {
                    item.set(VehicleType.VEHICLE_TYPE, vehicleType)
                            .set(BaseFields.CUSTOM_TYPE, customType);

                    if (!item.containsKey(BaseFields.tid) && item.containsKey(BaseFields.UPLOAD_TIME)) {
                        item.set(BaseFields.tid, item.get(BaseFields.UPLOAD_TIME));
                    }
                    item.set(BaseFields.PARTITION_KEY, DateUtil.format(DateUtil.date(item.getLong(BaseFields.tid)), DatePattern.NORM_DATE_PATTERN));
                    if (!item.containsKey("longitude") || !item.containsKey("latitude")) {
                        item.set("latitude", BaseFields.defaultVal);
                        item.set("longitude", BaseFields.defaultVal);
                    }
                    importanKeySupplement(item);
                    return item;
                });

        return mapStream;
    }

    /**
     * 格式化 字段名称 和补充
     *
     * @param source
     * @return
     */
    public JSONObject formatField(JSONObject source) {
        source = standardized(source);
        ReplenishField.replenish(source);
        return source;
    }

    /**
     * 字段标准化
     *
     * @param source
     * @return
     */
    protected JSONObject standardized(JSONObject source) {
        List<DwdStandardField> list = getStandards(customType);
        Map<String, String> filedsMap = StandardFieldTrans.getFieldMap(vehicleType, customType);
        // 注意 fieldMap 可能为空
        JSONObject result = JSONUtil.createObj();
        list.forEach(item -> {
            String filedName = item.getFieldName();
            String val = BaseFields.defaultVal;
            if (source.containsKey(filedName)) {
                val = source.getStr(filedName);
                if (!StrUtil.DASHED.equals(val)) {
                    val = rangeJudge(val, item);
                } else {
                    val = BaseFields.defaultVal;
                }
            } else {
                if (filedsMap.size() > 0) {
                    String originalKey = filedsMap.get(filedName);
                    if (source.containsKey(originalKey)) {
                        val = source.getStr(originalKey);
                        if (!StrUtil.DASHED.equals(val)) {
                            val = rangeJudge(val, item);
                        } else {
                            val = BaseFields.defaultVal;
                        }
                    }
                }
            }
            result.set(filedName, val);
        });
        // 是否再放开一个新的空方法供子类来添加自己的字段
        additionalFieldHandle(result);
        result.set(BaseFields.PARTITION_KEY, source.getStr(BaseFields.PARTITION_KEY));
        return result;
    }

    protected List<DwdStandardField> getStandards(String customType) {
        List<DwdStandardField> list = StandardPoolField.getStandardField(customType);
        addStandards(list);
        return list;
    }

    /**
     * 对标准字段补充
     *
     * @param list
     */
    protected void addStandards(List<DwdStandardField> list) {

    }

    protected void additionalFieldHandle(JSONObject source) {

    }


    /**
     * can矩阵范围校验
     *
     * @param val
     * @param field
     * @return
     */
    private String rangeJudge(String val, DwdStandardField field) {
        if (field.getMaxVal() != null) {
            // 比较
            if (Double.valueOf(val) > field.getMaxVal()) {
                return BaseFields.defaultVal;
            }
        }
        if (field.getMinVal() != null) {
            if (Double.valueOf(val) < field.getMinVal()) {
                return BaseFields.defaultVal;
            }
        }
        return val;
    }

    /**
     * 如果可以重写改方法，灵活消费策略
     *
     * @return
     */
//    public FlinkKafkaConsumer<String> kafkaConsumer() {
//        return FlinkKafkaUtil.getConsumerGroupOffsets(groupId, topic);
//    }
    public Boolean mileageFilter(JSONObject source) {
        if (source.containsKey(mileage) && StrUtil.isNotBlank(source.getStr(mileage))) {
            if (!BaseFields.defaultVal.equals(source.getStr(mileage)) && !StrUtil.DASHED.equals(source.getStr(mileage))) {
                if (Double.valueOf(DataPrecisionConfigEnum.ODOMETER.getValue()) > source.getDouble(mileage)) {
                    return false;
                }
            }
        }
        return true;
    }

    /**
     * 关键字段的补充处理，调用dwd的统一处理方法，可子类重写
     *
     * @param source
     */
    public void importanKeySupplement(JSONObject source) {
        DwdCheck.getDwdCheck()
                .setVehicleType(vehicleType)
                .check(source);
    }


}
